package cn.com.sparkle.paxos;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.event.ConfigureEvent;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.MasterChangePosEvent;
import cn.com.sparkle.paxos.event.NodeStateChangeEvent;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionId;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
//import cn.com.sparkle.paxos.model.ElectionId;
//import cn.com.sparkle.paxos.model.Id;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.util.IdComparator;

/**
 * i adopt mvcc to enhance the performance in concurrency.So the new instance of
 * NodesCollection may be created a lot of times.
 * 
 * @author nanqi
 * 
 */
public class ClusterState {
	private final static Logger logger = Logger.getLogger(ClusterState.class);

	private Configuration conf;

	// cur master
	private ElectionId lastElectionId;
	private volatile NodesCollection senators;
	private volatile NodesCollection followers;

	private EventsManager eventsManager;

	private SelfState selfState;

	private ReentrantLock nodeLock = new ReentrantLock();

	public ClusterState(EventsManager eventsManager, Configuration conf) {
		this.conf = conf;
		this.selfState = new SelfState(conf);
		this.eventsManager = eventsManager;
		this.lastElectionId = ElectionId.newBuilder()
				.setId(Id.newBuilder().setAddress("").setIncreaseId(-1))
				.setVersion(conf.getVersion()).build();
		// new ElectionId(-1, "", conf.getVersion());
		HashMap<String, NodeState> temp = new HashMap<String, NodeState>();
		for (ConfigNode node : conf.getSenators()) {
			String address = node.getIp() + ":" + node.getPort();
			NodeState ns = new NodeState(address);
			temp.put(address, ns);
		}
		senators = new NodesCollection(conf.getSenators(),
				new HashSet<NetNode>(), temp);
		temp = new HashMap<String, NodeState>();
		for (ConfigNode node : conf.getFollowers()) {
			String address = node.getIp() + ":" + node.getPort();
			NodeState ns = new NodeState(address);
			temp.put(address, ns);
		}
		followers = new NodesCollection(conf.getFollowers(),
				new HashSet<NetNode>(), temp);

		eventsManager.registerEvent(new ConfigureEvent() {
			@Override
			public void senatorsChange(Set<ConfigNode> newSenators, long version) {
				try {
					nodeLock.lock();
					HashSet<NetNode> activeNodes = new HashSet<NetNode>();
					for (NetNode nNode : senators.getActiveNodes()) {
						if (newSenators.contains(ConfigNode.parseNode(nNode
								.getNodeState().getAddress()))) {
							activeNodes.add(nNode);
						}
					}
					HashMap<String, NodeState> temp = new HashMap<String, NodeState>();
					for (ConfigNode node : newSenators) {
						String address = node.getIp() + ":" + node.getPort();

						if (senators.getNodeStates().containsKey(address)) {
							temp.put(address,
									senators.getNodeStates().get(address));
						} else {
							temp.put(address, new NodeState(address));
						}
					}
					senators = new NodesCollection(newSenators, activeNodes,
							temp);
				} finally {
					nodeLock.unlock();
				}

			}

			@Override
			public void followersChange(Set<ConfigNode> newFollowers) {
				try {
					nodeLock.lock();
					HashSet<NetNode> activeNodes = new HashSet<NetNode>();
					for (NetNode nNode : followers.getActiveNodes()) {
						if (newFollowers.contains(ConfigNode.parseNode(nNode
								.getNodeState().getAddress()))) {
							activeNodes.add(nNode);
						}
					}
					HashMap<String, NodeState> temp = new HashMap<String, NodeState>();
					for (ConfigNode node : newFollowers) {
						String address = node.getIp() + ":" + node.getPort();
						if (followers.getNodeStates().containsKey(address)) {
							temp.put(address,
									followers.getNodeStates().get(address));
						} else {
							temp.put(address, new NodeState(address));
						}
					}
					followers = new NodesCollection(newFollowers, activeNodes,
							temp);
				} finally {
					nodeLock.unlock();
				}
			}
		});
		eventsManager.registerEvent(new NodeStateChangeEvent() {
			@Override
			public void openConnect(NetNode nNode) {
				try {
					nodeLock.lock();
					ConfigNode cNode = ConfigNode.parseNode(nNode
							.getNodeState().getAddress());
					if (senators.getNodeMembers().contains(cNode)) {
						HashSet<NetNode> newActiveNodes = new HashSet<NetNode>();
						newActiveNodes.addAll(senators.getActiveNodes());
						newActiveNodes.add(nNode);
						senators = new NodesCollection(senators
								.getNodeMembers(), newActiveNodes, senators
								.getNodeStates());
					} else if (followers.getNodeMembers().contains(cNode)) {
						HashSet<NetNode> newActiveNodes = new HashSet<NetNode>();
						newActiveNodes.addAll(followers.getActiveNodes());
						newActiveNodes.add(nNode);
						followers = new NodesCollection(followers
								.getNodeMembers(), newActiveNodes, followers
								.getNodeStates());
					}
					logger.debug("add a node " + cNode.toString());
				} finally {
					nodeLock.unlock();
				}
			}

			@Override
			public void loseConnect(NetNode nNode) {
				try {
					nodeLock.lock();
					String address = nNode.getNodeState().getAddress();
					ConfigNode cNode = ConfigNode.parseNode(address);

					if (senators.getNodeMembers().contains(cNode)) {
						// mvcc
						HashMap<String, NodeState> newNodeState = new HashMap<String, NodeState>();
						newNodeState.putAll(senators.getNodeStates());
						newNodeState.put(address, new NodeState(address));
						HashSet<NetNode> newActiveNodes = new HashSet<NetNode>();
						newActiveNodes.addAll(senators.getActiveNodes());
						newActiveNodes.remove(nNode);
						senators = new NodesCollection(senators
								.getNodeMembers(), newActiveNodes, newNodeState);
					} else if (followers.getNodeMembers().contains(cNode)) {
						// mvcc
						HashMap<String, NodeState> newNodeState = new HashMap<String, NodeState>();
						newNodeState.putAll(followers.getNodeStates());
						newNodeState.put(address, new NodeState(address));
						HashSet<NetNode> newActiveNodes = new HashSet<NetNode>();
						newActiveNodes.addAll(followers.getActiveNodes());
						newActiveNodes.remove(nNode);
						followers = new NodesCollection(followers
								.getNodeMembers(), newActiveNodes, newNodeState);
					}

					logger.debug("lost a node " + address);
				} finally {
					nodeLock.unlock();
				}

			}

			@Override
			public void beatHeart(NodeState nState) {
				try {
					nodeLock.lock();
					ConfigNode cNode = ConfigNode.parseNode(nState.getAddress());
					if (senators.getNodeMembers().contains(cNode)) {
						HashMap<String, NodeState> newNodeState = new HashMap<String, NodeState>();
						newNodeState.putAll(senators.getNodeStates());
						newNodeState.put(nState.getAddress(), nState);
						senators = new NodesCollection(senators
								.getNodeMembers(), senators.getActiveNodes(),
								newNodeState);
					} else if (followers.getNodeMembers().contains(cNode)) {
						HashMap<String, NodeState> newNodeState = new HashMap<String, NodeState>();
						newNodeState.putAll(followers.getNodeStates());
						newNodeState.put(nState.getAddress(), nState);
						followers = new NodesCollection(followers
								.getNodeMembers(), followers.getActiveNodes(),
								newNodeState);
					}
				} finally {
					nodeLock.unlock();
				}
			}
		});
	}

	private ReentrantLock electionIdLock = new ReentrantLock();

	public ElectionId getLastElectionId() {
		try {
			electionIdLock.lock();
			return lastElectionId;
		} finally {
			electionIdLock.unlock();
		}
	}

	public boolean changeLastElectionId(ElectionId lastElectionId) {
		try {
			electionIdLock.lock();

			ElectionId id = selfState.casElectionPrepareId(lastElectionId);
			if (id == lastElectionId &&
			// lastElectionId.compareTo(this.lastElectionId) == 1) {
					IdComparator.getInstance().compare(lastElectionId.getId(),
							this.lastElectionId.getId()) == 1) {
				logger.debug("master change:"
						+ lastElectionId.getId().getIncreaseId() + "  "
						+ lastElectionId.getId().getAddress());
				if (this.lastElectionId.getId().getAddress()
						.equals(conf.getSelfAddress())) {
					MasterChangePosEvent.doLostPosEvent(eventsManager);
				} else if (lastElectionId.getId().getAddress()
						.equals(conf.getSelfAddress())) {
					MasterChangePosEvent.doGetMasterPosEvent(eventsManager);
				}
				this.lastElectionId = lastElectionId;
				MasterChangePosEvent.doMasterChangeEvent(eventsManager,
						lastElectionId.getId().getAddress());
				return true;
			}
			return false;
		} finally {
			electionIdLock.unlock();
		}
	}

	public void lostAuthorizationOfMaster() {
		try {
			electionIdLock.lock();
			if (lastElectionId.getId().getAddress()
					.equals(conf.getSelfAddress())) {
				lastElectionId = ElectionId
						.newBuilder()
						.setId(Id.newBuilder().setAddress("").setIncreaseId(-1))
						.setVersion(lastElectionId.getVersion()).build();
				// lastElectionId = new ElectionId(-1, "",
				// lastElectionId.getVersion());
				MasterChangePosEvent.doLostPosEvent(eventsManager);
			}
		} finally {
			electionIdLock.unlock();
		}
	}

	public SelfState getSelfState() {
		return selfState;
	}

	public NodesCollection getSenators() {
		return senators;
	}

	public NodesCollection getFollowers() {
		return followers;
	}

}
